package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.euler.commons.web.JsonMapperBuilder;
import com.rrd.dw.mr.gzip.GzipMR;
import com.rrd.dw.utils.HadoopUtils;
import com.ucredit.babel.common.kafka.PostRecord;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.alibaba2.AliPayRecord;
import com.ucredit.babel.common.model.alibaba2.AliTaobaoRecord;
import com.ucredit.babel.common.model.app.ApplistRecords;
import com.ucredit.babel.common.model.behavior.UnloginUserBehaviorRecord;
import com.ucredit.babel.common.model.behavior.UserBehaviorRecord;
import com.ucredit.babel.common.model.call.CallRecord;
import com.ucredit.babel.common.model.dhb.DhbItagInfoRecords;
import com.ucredit.babel.common.model.nfcs.NfcsReport;
import com.ucredit.babel.common.model.pcac.PcacRiskInfo;
import com.ucredit.babel.common.model.qcloud.QcloudAntiFraudRecords;
import com.ucredit.babel.common.model.third.jiguang.JiGuangAddressRecord;
import com.ucredit.babel.common.model.third.jiguang.JiGuangFactorRecord;
import com.ucredit.babel.common.model.third.unionpaynew.UnionPayRiskReportRecord;
import com.ucredit.babel.common.model.third.xinyan.XinyanBlackRecord;
import com.ucredit.babel.common.model.third.xinyan.XinyanRadarRecord;
import com.ucredit.babel.common.model.third.xinyan.XinyanWhiteRecord;
import com.ucredit.babel.common.model.tianjiscore.TianjiBasicScoreRecord;
import com.ucredit.babel.common.model.unionpaysmart.UnionPaySmartPersonal;
import com.ucredit.babel.common.model.xwbank.XwbankApproval;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;

public class CallRecordsMR extends Configured implements Tool {

    public static final String CRT001 = "\001";
    public static final Pattern CRT001_PATTERN = Pattern.compile("\001");
    private static final List<ParseBase> ParserList = getParseList();

    public static class CallMap extends Mapper<LongWritable, Text, Text, Text> {
        private Set<String> appSet = null;
        private Map<String, List<ParseBase>> parserListMap = new HashMap<>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String appType = context.getConfiguration().get("etl.param.app.type", "");
            if (appType != null && !appType.equals("")) {
                appSet = new HashSet<>(Arrays.asList(appType.split(",")));
            }
            // parser list
            setParserMap();

        }

        private void setParserMap() {
            for (ParseBase parser : getParseList()) {
                List<ParseBase> parsers;
                if (parserListMap.containsKey(parser.getType())) {
                    parsers = parserListMap.get(parser.getType());
                } else {
                    parsers = new ArrayList<>();
                }
                parsers.add(parser);
                parserListMap.put(parser.getType(), parsers);
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            if (value == null || value.toString().equals("")) {
                return;
            }
            try {
                JSONObject jsonObj = JSON.parseObject(value.toString());
                String type = jsonObj.getString("type");
                if (type == null || type.equals("CREDIT_UPLOAD")) {
                    return;
                }
                if (appSet != null && !appSet.contains(jsonObj.getString("type"))) {
                    return;
                }
                List<ParseBase> parsers = parserListMap.get(jsonObj.getString("type"));
                if (parsers != null && parsers.size() > 0) {
                    PostRecord record = JsonMapperBuilder.build().readValue(value.toString(), PostRecord.class);
                    BaseModel<?> model = (BaseModel<?>) record.getRecord();
                    if (record.getAction().equals("PUT") || record.getAction().equals("NOTIFY")) {
                        for (ParseBase parser : parsers) {
                            parseValue(model, parser, context);
                        }
                    }
                } else {
                    System.out.println("miss type:|" + jsonObj.getString("type") + "|");
                }
            } catch (Exception e) {
                context.getCounter("error", "ClassCastException").increment(1);
                System.out.println("文本:|" + value.toString() + "|");
                e.printStackTrace();
//				throw e;
            }

        }

        protected void parseValue(BaseModel<?> model, ParseBase parser, Context context) throws IOException, InterruptedException {
            String successful = "1";
            if (model != null) {
                if (parser.parseJob(context, model, successful)) {
                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType()).increment(1);
                } else {
//                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType() + ".false").increment(1);
                }
            } else {
                context.getCounter("error", "user." + parser.getType() + parser.getSubType() + ".null").increment(1);
            }
        }
    }

    public static class CallRecordReducer extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private NullWritable key2 = NullWritable.get();
        private String dt = null;
        private Map<String, String> map = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<Text, Text>(context);
            Configuration conf = context.getConfiguration();
            this.dt = conf.get("etl.param.dt", "2018-04-27");
            this.map = getOutPath(dt);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            this.mos.close();
        }

        private String keystr = null;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            keystr = key.toString().split(",")[0];
            System.out.println("REDUCE KEY:" + keystr + "|" + this.map.get(keystr) + "|" + key.toString());
            for (Text val : values) {
                if (this.map.get(keystr) == null) {
                    System.out.println("ERROR KEY:|" + keystr);
                    context.getCounter("reduce", "reduce.null").increment(1);
//					System.out.println(keystr+"==>"+);
                } else {
                    context.getCounter("reduce", "reduce.success").increment(1);
                    this.mos.write("textinfo", key2, val, this.map.get(keystr));
                }
            }

        }
    }

    private static List<ParseBase> getParseList() {
        ArrayList<ParseBase> parsers = new ArrayList<>();
        parsers.add(new BaihangBlackListParse());
        parsers.add(new BairongUserReportParse());
        parsers.add(new LongjingScoreParse());
        parsers.add(new PhoneHelpUMParse());
        parsers.add(new TongdunZhiXinFenParse());
        parsers.add(new PHoneHelpPScoreParse());
        parsers.add(new BaiweiScoreParse());
        parsers.add(new BaiweiFraudScoreParse());
        parsers.add(new QianYunScoreParse());
        parsers.add(new BairongRecordParse());
        parsers.add(new BaihangC1Parse());
        parsers.add(new NifaQueryParse());
        parsers.add(new NifaQueryDetailParse());
        parsers.add(new UmengCreditScoreParse());
        parsers.add(new UmengCareerFeatureParse());
        parsers.add(new NewPhoneCallRecordParse());
        parsers.add(new NewSmsReportParse());
        parsers.add(new OneConnectOffenlendParse());
        parsers.add(new OneConnectScenesriskParse());
        parsers.add(new OneConnectRisknewsParse());
        parsers.add(new OneConnectFraudnewsParse());
        parsers.add(new ZitacrDishonesty4Parse());
        parsers.add(new YpcreditParse());
        parsers.add(new TianjiScoreV7Parse());
        parsers.add(new TianjiAntifraudV2Parse());
        parsers.add(new IceKreditParse());
        parsers.add(new JinDunParse());
        parsers.add(new ShieldScoreParse());
        parsers.add(new TianjiScoreV6Parse());
        parsers.add(new DataVisorParse());
        parsers.add(new BairongLimitConsumptionParse());
        parsers.add(new QcloudAntiFraudConsumptionParse());
        parsers.add(new TianjiActionParse());
        parsers.add(new TianjiActionV2Parse());
        parsers.add(new TianjiLocationV2Parse());
//        parsers.add(new TongdunRiskReportParse());
//        parsers.add(new TongdunRiskHitRulesParse());
//        parsers.add(new TongdunRiskDetailConditionsInfoParse());
        parsers.add(new FaYanYuanCaseParse());
        parsers.add(new FaYanYuanDishonestyParse());
        parsers.add(new FinanceGpAppParse());
        parsers.add(new PhonehelpOnlinestatusParse());
        parsers.add(new PhonehelpTingjiParse());
        parsers.add(new TianjiAntifraudParse());
        parsers.add(new TianjiLocationParse());
        parsers.add(new XinYanBlackListParse());
        parsers.add(new SmartDataParse());
        parsers.add(new HjxhInenterParse());
        parsers.add(new HjxhWholeInParse());
        parsers.add(new JiTuiMobileOnlineParse());
        parsers.add(new JiTuiMobileStatusParse());
        parsers.add(new ZhongQingParse());
        parsers.add(new XinDeGrefListParse());
        parsers.add(new JdPgBlankListParse());
        parsers.add(new JdPgScoreParse());
        parsers.add(new DuXiaoManCreditScoreParse());
        parsers.add(new DuXiaoManMultiloansParse());
        parsers.add(new WitskyMobileDnParse());
        parsers.add(new MobileCreditScoreY01Parse());
        parsers.add(new NewPhoneBookParse());
        parsers.add(new UnionpayAdvisorsFtParse());
        parsers.add(new UnionpayAdvisorsUpParse());
        parsers.add(new YootungConsumptiveForecastParse());
        parsers.add(new FaceidOCRParse());
        parsers.add(new TencentCloudOCRParse());
        parsers.add(new TxCreditMainParse());
        parsers.add(new TxCreditMainAdditionalParse());
        parsers.add(new TxCreditSiFaParse());
        parsers.add(new TxCreditShiXinParse());
        parsers.add(new TxCreditFengXianParse());
        parsers.add(new TxCreditGuQuanParse());
        parsers.add(new HeiTongAntiFraudScoreBParse());
        parsers.add(new XtmConsumptionScoreParse());
        parsers.add(new TencentAntifraudVipParse());
        parsers.add(new BaiRongPortraitRecordParse());
        parsers.add(new BairongPortraitPreds10RecordParse());
        parsers.add(new SuanHuaRiskScoreRecordParse());
        parsers.add(new SuanHuaCreditScoreRecordParse());
        parsers.add(new BaiRongSpecialListRecordParse());
        parsers.add(new BaiRongStragyLoanRecordParse());
        parsers.add(new SuanHuaRiskIdentRecordParse());
        parsers.add(new YpcreditAssessmentParse());
        parsers.add(new BaiRongPortraitCreditModelParse());
        parsers.add(new BaiRongUserCustomizeCreditScore3Parse());
        parsers.add(new BaiRongUserCustomizeCreditScore4Parse());
        parsers.add(new BaiRongUserCustomizeCreditScore5Parse());
        parsers.add(new BaiweiScorePlusParse());
        parsers.add(new BaihangQ1RecordParse());
        parsers.add(new DuXiaoManFraudFactorParse());
        parsers.add(new DuXiaoManRiskListParse());
        parsers.add(new QCloudAccessRiskRecordParse());
        parsers.add(new HzphfinCreditScoreB1RecordParse());
        parsers.add(new HzphfinFraudScoreT3RecordParse());
        parsers.add(new HjxhISPThreeParse());
        parsers.add(new JiTuiMobileFactorThreeParse());
        parsers.add(new TcreditUnitedScoreParse());
        parsers.add(new BaihangAirCreditParse());
        parsers.add(new TianchuangPayScoreParse());
        parsers.add(new TpCreditSinanRecordParse());
        parsers.add(new ShangHaiNfcsReportAddress());
        parsers.add(new ShangHaiNfcsReportApplication());
        parsers.add(new ShangHaiNfcsReportContact());
        parsers.add(new ShangHaiNfcsReportMain());
        parsers.add(new ShangHaiNfcsReportQuerryRecord());
        parsers.add(new ShangHaiNfcsReportTransaction());
        parsers.add(new HfgPhoneDeviceInfo());
        parsers.add(new HfgGpsInfo());
        parsers.add(new YouMengFraudScoreParse());
        parsers.add(new YouMengFraudScoreV2Parse());
        parsers.add(new JingDongReimbursementAbilityParse());
        parsers.add(new JingDongAccidentModelSectionParse());
        parsers.add(new BingJianDayPupilParse());
        parsers.add(new BingJianEyeParse());
        parsers.add(new BiLeiZhenTuZhi04Parse());
        parsers.add(new BiLeiZhenTuZhi07Parse());
        parsers.add(new BiLeiHitListParse());
        parsers.add(new XinYiSouBlackListParse());
        parsers.add(new XinYiSouZiYu3ScoreParse());
        parsers.add(new XinYiSouWanXiang2ScoreParse());
        parsers.add(new XinYiSouFeiCui2ScoreParse());
        parsers.add(new XinYiSouLanYu2ScoreParse());
        parsers.add(new WeichengRecordParse());
        parsers.add(new BaihangSMEIS002Parse());
        parsers.add(new BairongPortraitScorelcashonRecordParse());
        parsers.add(new PhonehelpCreditScoreRecordParse());
        parsers.add(new QiChaChaShixinListRecordParse());
        parsers.add(new QiChaChaSumptuaryListRecordParse());
        parsers.add(new QiChaChaZhixingListRecordParse());
        parsers.add(new TanZhiV2RecordParse());
        parsers.add(new PuDaoCustomerLabelRecordParse());
        parsers.add(new PuDaoMobileVerificationRecordParse());
        parsers.add(new YunXingScoreRecordParse());
        parsers.add(new YinLianBusinessBlackListRecordParse());
        parsers.add(new BaihangSMEIS01CRecordParse());
        parsers.add(new CalendarInfoRecordParse());
        parsers.add(new HfqPlistRecordParse());
        parsers.add(new HfqSmsReportRecordParse());
        parsers.add(new HfqPhoneBookRecordParse());
        parsers.add(new HfqCalendarRecordParse());
        parsers.add(new HfqPhoneCallRecordParse());

        parsers.add(new BairongInterestRatePreferenceParse());
        parsers.add(new BairongUserCustomizeCreditScore10Parse());
        parsers.add(new BairongUserCustomizeCreditScore11Parse());
        parsers.add(new BairongUserCustomizeCreditScore12Parse());
        parsers.add(new BairongUserCustomizeCreditScore13Parse());
        parsers.add(new FundsCreditRecordParse());
        parsers.add(new heikaFeatureRecordParse());
        parsers.add(new TongdunWuLiuHuaXiangParse());
        parsers.add(new PuDaoLianYangScoreAParse());
        parsers.add(new PuDaoLianYangScoreBParse());
        parsers.add(new PuDaoFederalLearningRiskModelParse());
        parsers.add(new PuDaoMultipleApplyParse());
        parsers.add(new PuDaoCurrentDayApplyParse());
        parsers.add(new BaihangPortraitParse());
        parsers.add(new BaihangRuleResultParse());
        parsers.add(new BaihangAntifraudprdParse());
        parsers.add(new BaihangRIDS0000Parse());
        parsers.add(new BaihangFRAI001AParse());
        parsers.add(new PuDaoTXLoanBehaviorUsualParse());
        parsers.add(new TongDunRiskV2Parse());
        parsers.add(new TongDunRiskHitDimV2Parse());
        parsers.add(new TongDunRiskConditionsV2Parse());
        parsers.add(new TongDunRiskHitRulesV2Parse());
        parsers.add(new BairongRiskRatingParse());
        parsers.add(new TcreditIntegrationParse());
        parsers.add(new CommunicateScoreParse());
        parsers.add(new HzphfinFraudScoreA1Parse());
        parsers.add(new IncomeStabilityScoreParse());
        parsers.add(new IncomeAbilityScoreParse());
        parsers.add(new IceCreditRateStratificationParse());
        parsers.add(new AntChainAntiFraudValueParse());
        parsers.add(new BaihangModelTagParse());
        parsers.add(new HfgUploadFraudParse());
        parsers.add(new HfgGpsInfoV2());
        parsers.add(new XinYiSouWanXiang1646Parse());
        parsers.add(new XinYiSouWanXiang16394Parse());
        parsers.add(new PuDaoByteLoanCreditScoreParse());
        parsers.add(new PuDaoTxyComprehensiveRiskLabelParse());
        parsers.add(new PuDaoTxySuspectedDefraudstersParse());
        parsers.add(new BairongBR0003814Parse());
        parsers.add(new BaihangPCAS0018Parse());
        parsers.add(new TongDunAddrBaseInfoParse());
        parsers.add(new TongDunAddrFrequencyParse());
        parsers.add(new PuDaoByteLoanCreditScoreZ1113Parse());
        parsers.add(new PuDaoJiGuangFintechNB2Parse());
        parsers.add(new PuDaoJiGuangFintechRS7Parse());
        parsers.add(new HfqSmsReportMonRecordParse());
        parsers.add(new BaihangTCBLScoreParse());
        parsers.add(new DuXiaoManCreditScoreV10Parse());
        parsers.add(new PuDaoFaHaiSiFaListParse());
        parsers.add(new PuDaoFaHaiSiFaDetailParse());
        parsers.add(new BaihangBHWY023AParse());
        return parsers;
    }

    private static Map<String, String> getOutPath(String dt) {
        Map<String, String> map = new HashMap<>();
        for (ParseBase parser : ParserList) {
            map.put(parser.getType() + parser.getSubType(), "/data/fdm/" + parser.getTableName() + "/dt=" + dt + "/");
        }
        return map;
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
        Job job = Job.getInstance(conf);
        String dt = conf.get("etl.param.dt", "2018-04-19");
        Map<String, String> map = getOutPath(dt);

        System.setProperty("HADOOP_USER_NAME", "hdfs");

        String input = conf.get("etl.param.input", "/data/hbase/call_records/" + dt);
        String output = conf.get("etl.param.output", "/tmp/bdm/call_records");
        String type = conf.get("etl.param.app.type", "");
        String delPaths = conf.get("etl.param.delete.paths");
        if (dt.equals("")) {
            System.out.println("请输入日期：etl.param.dt");
            return -1;
        }
        if (output.equals("")) {
            System.out.println("请输入输出路径：etl.param.output");
            return -1;
        }
        if (!type.equals("")) {
            System.out.println("此次单独输出：" + type);
            System.out.println("此次删除目录标识：" + delPaths);
        }
        System.out.println("pdl输入路径###:" + input);
        System.out.println("输出路径:" + output);
        String isFilterTmp = conf.get("param.is.filter.tmp", "false");
        HadoopUtils.delete(conf, new Path(output));
        List<String> delList = null;
        if (delPaths != null) {
            delList = Arrays.asList(delPaths.split(","));
        }
        for (String s : map.keySet()) {
            if (delList != null && delList.size() > 0) {
                String ss = s.split("\\.")[0];
                if (delList.contains(ss)) {
                    HadoopUtils.delete(conf, new Path(map.get(s)));
                }
            } else {
                HadoopUtils.delete(conf, new Path(map.get(s)));
            }
        }
        HadoopUtils.delete(conf, new Path("/tmp/data/userbehavior"));
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        if (isFilterTmp.equals("true")) {
            FileSystem fs = FileSystem.get(new Configuration());
            FileStatus[] status = fs.globStatus(new Path(input + "/*"), new RegexExcludePathFilter(".*\\.data(?<!\\.tmp)$"));
            Path[] listedPaths = FileUtil.stat2Paths(status);
            if (listedPaths == null || listedPaths.length <= 0) {
                System.out.println("输入路径除去.tmp结尾的文件，没有其他文件！");
                System.exit(0);
            }
            TextInputFormat.setInputPaths(job, listedPaths);
        } else {
            TextInputFormat.setInputPaths(job, input);
        }
        TextOutputFormat.setOutputPath(job, new Path(output));
        job.setJarByClass(this.getClass());
        job.setMapperClass(CallMap.class);
        job.setReducerClass(CallRecordReducer.class);
        MultipleOutputs.addNamedOutput(job, "textinfo", TextOutputFormat.class, Text.class, Text.class);

        int stat = job.waitForCompletion(true) ? 0 : 1;
        System.out.println("異常次數：" + job.getCounters().getGroup("error").findCounter("ClassCastException").getValue());
        for (ParseBase parser : ParserList) {
            System.out.println(parser.getComment() + parser.getSubType() + ":" + job.getCounters().getGroup("counter").findCounter("user." + parser.getType() + parser.getSubType()).getValue());
            System.out.println(parser.getComment() + parser.getSubType() + "查询is Null次数：" + job.getCounters().getGroup("error").findCounter("user." + parser.getType() + parser.getSubType() + ".null").getValue());
        }

        System.out.println("reduce.null次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.null").getValue());
        System.out.println("reduce.success次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.success").getValue());
        System.out.println("user.unionPayRiskReportRecord.record.null次数：" + job.getCounters().getGroup("counter").findCounter("user.unionPayRiskReportRecord.record.null").getValue());
        System.out.println("user.unionPayRiskReportRecord.record.not.null次数：" + job.getCounters().getGroup("counter").findCounter("user.unionPayRiskReportRecord.record.not.null").getValue());
        System.out.println("user.unionPayRiskReportRecord.put.null次数：" + job.getCounters().getGroup("error").findCounter("user.unionPayRiskReportRecord.put.null").getValue());

        return stat;
    }

    private class RegexExcludePathFilter implements PathFilter {
        private final String regex;

        public RegexExcludePathFilter(String regex) {
            this.regex = regex;
        }

        @Override
        public boolean accept(Path path) {
            System.out.println("forpath:" + path + "," + path.toString().matches(regex));
            return path.toString().matches(regex);
        }
    }

    public static void main(String[] args) {
        int n = 0;
        try {
            n = ToolRunner.run(new CallRecordsMR(), args);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.exit(n);
    }
}
